package com.example.rocket.producer;

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.example.rocket.config.NbMqProperties;
import com.example.rocket.core.NbCallback;
import com.example.rocket.core.NbSendResult;
import com.example.rocket.utils.JSONUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.Date;
import java.util.Properties;

/**
 * @description: OnsMQProducer - 采用 ons-client
 * @author TAO
 * @date 2021/12/21 3:29 下午
 */
@Slf4j
public class OnsMqProducer implements NbMqProducer {

    private static Producer producer = null;

    public OnsMqProducer(NbMqProperties properties) {
        log.info("↓↓↓↓----↓↓↓↓----↓↓↓↓----↓↓↓↓");
        log.warn("-->当前环境启用===> O N S <--");
        log.info("↑↑↑↑----↑↑↑↑----↑↑↑↑----↑↑↑↑");
        Properties mqProperties = new Properties();
        mqProperties.put(PropertyKeyConst.GROUP_ID, properties.getProducer().getProducerId());
        mqProperties.put(PropertyKeyConst.AccessKey, properties.getAccessKey());
        mqProperties.put(PropertyKeyConst.SecretKey, properties.getSecretKey());
        mqProperties.put(PropertyKeyConst.NAMESRV_ADDR, properties.getNameServer());

        producer = ONSFactory.createProducer(mqProperties);
        producer.start();
    }


    /**
     * 同步消息
     * @param topic
     * @param tag
     * @param payload
     * @return
     */
    @Override
    public String syncSend(String topic, String tag, Object payload) {
        try {
            Message message = new Message(topic, tag, JSONUtils.writeAsBytes(payload));
            return producer.send(message).getMessageId();
        } catch (ONSClientException e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            log.info("===>" + new Date() + " 发送MQ消息失败！主题是：" + topic);
            throw new RuntimeException(e);
        }
    }

    /**
     * 异步消息
     * @param topic
     * @param tag
     * @param payload
     * @param sendCallback
     */
    @Override
    public void asyncSend(String topic, String tag, Object payload, NbCallback sendCallback) {
        try {
            Message message = new Message(topic, tag, JSONUtils.writeAsBytes(payload));
            producer.sendAsync(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    sendCallback.onSuccess(sendResult.getMessageId());
                }

                @Override
                public void onException(OnExceptionContext context) {
                    sendCallback.onException(context.getException());
                }
            });
        } catch (ONSClientException e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            log.info("===>" + new Date() + " Send mq message failed! Topic is:" + topic);
            throw new RuntimeException(e);
        }
    }

    /**
     * 单向消息
     * @param topic
     * @param tag
     * @param payload
     */
    @Override
    public void sendOneWay(String topic, String tag, Object payload) {
        try {
            Message message = new Message(topic, tag, JSONUtils.writeAsBytes(payload));
            producer.sendOneway(message);
        } catch (ONSClientException e) {
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            log.info("===>" + new Date() + " Send mq message failed! Topic is:" + topic);
            throw new RuntimeException(e);
        }
    }

    @Override
    public NbSendResult batchSend(String topic, String tag, Collection<Object> messages) {


        return null;
    }

    /**
     * 同步延迟消息
     *
     * @param topic
     * @param tag
     * @param message
     * @param timeout
     * @param delayLevel
     * @return
     */

    //private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";

     private int[] fixedDelayLevel =new int[]{1000,5000,10000,30000,60000,120000,180000,240000,300000,
            360000,420000,480000,540000,600000,1200000,1800000,3600000,7200000};

    @Override
    public NbSendResult syncSend(String topic, String tag, Object message, long timeout, int delayLevel) {
        if (delayLevel<1||delayLevel>fixedDelayLevel.length){
            log.info("===>" + new Date() + "当前Topic延迟时间异常" + delayLevel);
            throw new IllegalArgumentException("当前Topic延迟时间异常");
        }
        Message onsMessage = createOnsMqMessage(topic, tag, message);
        onsMessage.setStartDeliverTime(System.currentTimeMillis() + fixedDelayLevel[delayLevel-1]);
        try {
            SendResult sendResult = producer.send(onsMessage);
            assert sendResult != null;
            log.info(new Date() + "发送MQ定时器消息成功！主题是：" + topic + "msgId is: " + sendResult.getMessageId());
            return new NbSendResult(sendResult);
        }catch (ONSClientException e){
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            log.info(new Date() + "发送 mq 消息失败。主题是：" + topic);
            e.printStackTrace();
            throw new IllegalArgumentException("发送 mq 消息失败。主题是：");
        }
    }

    /**
     * 异步延迟消息
     *  @param topic
     * @param tag
     * @param message
     * @param timeout
     * @param delayLevel
     * @param sendCallback
     */
    @Override
    public void asyncSend(String topic, String tag, Object message, long timeout, int delayLevel, NbCallback sendCallback) {
        if (delayLevel<1||delayLevel>fixedDelayLevel.length){
            log.info("===>" + new Date() + "当前Topic延迟时间异常" + delayLevel);
            throw new IllegalArgumentException("当前Topic延迟时间异常");
        }
        Message onsMessage = createOnsMqMessage(topic, tag, message);
        onsMessage.setStartDeliverTime(System.currentTimeMillis() + fixedDelayLevel[delayLevel-1]);
        try {
            producer.sendAsync(onsMessage,new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info(new Date() + "发送MQ定时器消息成功！主题是：" + topic + "msgId is: " + sendResult.getMessageId());
                    sendCallback.onSuccess(sendResult.getMessageId());
                }

                @Override
                public void onException(OnExceptionContext context) {
                    sendCallback.onException(context.getException());
                }
            });

        }catch (ONSClientException e){
            // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理
            log.info(new Date() + "发送 mq 消息失败。主题是：" + topic);
            e.printStackTrace();
            throw new IllegalArgumentException("发送 mq 消息失败。主题是：");
        }
    }

    public Message createOnsMqMessage(String topic, String tag, Object msg) {
        return new Message(topic, tag, JSONUtils.writeAsBytes(msg));
    }





}
